package com.gitee.etl.process.bean

import com.gitee.utils.KafkaProp
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011
import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper

trait BeanETL[T] {

  /**
    * 从kafka中读取数据，传递返回的数据类型
    *
    * @param topic
    * @return
    */
  def getKafkaDataStream(topic: String): DataStream[T];

  /**
    * 处理数据的接口
    */
  def process(): Unit;

  /**
    * 发送数据到kafka
    * @param topic
    * @return
    */
  def kafkaProducer(topic: String): FlinkKafkaProducer011[String] = {
    new FlinkKafkaProducer011[String](topic,new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()),KafkaProp.getProperties())
  }

}
